package com.kl.rabbitmqconsumer.config;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 手动确认消息监听类
 * 接口顾名思义：信道感知消息
 */
@Component
public class MyAckReceiver implements ChannelAwareMessageListener {

    /**
     * 感知消息并处理
     */
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        System.out.println("=======进入监听回调函数=======");
        // 消息的唯一标识
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            // 因为传递消息的时候用的map传递,所以将Map从Message内取出需要做些处理
            String msg = message.toString();
            // 可以点进Message里面看源码,单引号直接的数据就是我们的map消息数据
            System.out.println("监听到的原始消息=" + msg);
            String[] msgArray = msg.split("'");
            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim());
            String messageId = msgMap.get("messageId");
            String messageData = msgMap.get("messageData");
            String createTime = msgMap.get("createTime");

            if ("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())) {
                System.out.println("1.消费的消息来自的队列名为：" + message.getMessageProperties().getConsumerQueue());
                System.out.println("2.消息成功消费到  messageId:" + messageId + "  messageData:" + messageData + "  createTime:" + createTime);
                System.out.println("3.执行TestDirectQueue中的消息的业务处理流程......");
                // 确认消息
                // 第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
                channel.basicAck(deliveryTag, false);
                System.out.println("4.手动确认TestDirectQueue队列的消息已被消费");
            }

            if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())) {
                System.out.println("1.消费的消息来自的队列名为：" + message.getMessageProperties().getConsumerQueue());
                System.out.println("2.消息成功消费到  messageId:" + messageId + "  messageData:" + messageData + "  createTime:" + createTime);
                System.out.println("3.执行fanout.A中的消息的业务处理流程......");
                channel.basicAck(deliveryTag, false);
                System.out.println("4.手动确认fanout.A队列的消息已被消费");
            }
            // 设置了TestDirectQueue", "fanout.A", "fanout.B三个队列的消息需要手动确认
            // fanout.C没有设置，所有默认为自动确认，当fanout消息发送后，由于fanout.B没有确认，在服务端状态为Unacked
        } catch (Exception e) {
            // 业务处理失败就拒接消息
            // 第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }

    // {key=value,key=value,key=value} 格式转换成map
    // {createTime=2021-08-30 22:15:27, messageId=8f631633-3b05-49df-9dcb-9c4d3a745d49, messageData=test message, hello!}
    private Map<String, String> mapStringToMap(String str) {
        // 将json字符串反序列化为泛型对象
        Map<String, String> map = JSONObject.parseObject(str, new TypeReference<Map<String, String>>() {
        });
        System.out.println("反序列化得到的消息内容 = " + map);
        return map;
    }
}